#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <errno.h>
#include <string.h>
#include <log.h>
#include <memcheck.h>
#include <compress.h>


HTTP_SVR_NS_BEGIN

gzip_thread::gzip_thread():_tid(-1)
{

}


gzip_thread::~gzip_thread()
{
    if(_status==GZIP_THD_RUNNING)
    {
        Kill_thread();
    }
    _tid=-1;
}

int gzip_thread::Create_thread()
{
    if(_status!=GZIP_THD_UNINIT)
    {
       return -1; 
    }
    _signal=GZIP_THD_RUN_SIG; 
    /*
    if(pthread_create(&_tid,NULL,Gzip_running,this)!=0)
    {
        log_error("create gzip_thd failed");
        return -1;
    }
    log_debug("create gzip_thd success ,thd_id is %d",(int)_tid);
    */
    return 0;
}


int gzip_thread::Kill_thread()
{
   _signal=GZIP_THD_STOP_SIG;
   if(pthread_join(_tid,NULL)!=0)
   {
        log_error("wait gzip_thread of id:%d quit failed",(int)_tid);
        if(pthread_cancel(_tid)<0)
        {
            log_error("kill gzip_thread :%d by force failed",(int)_tid);
            return -1;
        }
        return 0;
   }
   return 0; 
}


void* gzip_thread::Gzip_running(void* arg)
{   
    if(arg==NULL)
        return NULL;
    gzip_thread* thd=(gzip_thread*)arg;
    vector<gzip_task>::iterator work_ite;
    while(1)
    {
        
        if(thd->_signal==GZIP_THD_PAUSE_SIG)
        {
            while(thd->_signal!=GZIP_THD_RESUME_SIG)
            {
                    usleep(1);
            }
        }
        
        if(thd->_signal== GZIP_THD_STOP_SIG)
        {
            thd->_status=GZIP_THD_STOPPED;
            return NULL;
        }

        if(thd->_waiting_task_queue==NULL)
        {
            usleep(1);
            continue;
        }
       
     //   sleep(2);
        thd->_working_task_queue=thd->_waiting_task_queue;
        
        
        for(work_ite=thd->_working_task_queue->begin();work_ite!=thd->_working_task_queue->end();work_ite++)
        {
            if(work_ite->_gzip_obj->get_terminate_flag()==1)
                continue;
            work_ite->_gzip_obj->add_extend_head_ext();        
        }
        
        thd->_waiting_task_queue=NULL;
            
        thd->_complete_task_queue.insert(thd->_complete_task_queue.end(),
                                        thd->_working_task_queue->begin(),thd->_working_task_queue->end());
   
        DELETE(thd->_working_task_queue);
        
        if(thd->_out_task_queue==NULL)
        {
            vector<gzip_task>* tmp_queue;
            NEW( vector<gzip_task> , tmp_queue);
            if( NULL == tmp_queue )
            {
                return NULL;
            }
            *tmp_queue=thd->_complete_task_queue;
            thd->_complete_task_queue.clear();
            thd->_out_task_queue=tmp_queue;
        }
    }
}



compress_unit* compress_unit::Instance()
{
    return CSingleton<compress_unit>::Instance();    
}


void    compress_unit::Destroy()
{
    CSingleton<compress_unit>::Destroy();
}

int compress_unit::Set_gzip_thread_num(int expect_num)
{
    int cur_thread_num=_thread_vec.size();
    if(expect_num>cur_thread_num)
    {
        for(int i=0;i<expect_num-cur_thread_num;i++)
        {
            gzip_thread* thread=NULL;
            NEW(gzip_thread,thread);
            if(thread==NULL)
            {
                log_error("create gzip_thread  failed,maybe memeory overflow");
                return -1;
            }
            if(thread->Create_thread()<0)
            {
                DELETE(thread);
                return -1;
            }
            _thread_vec.push_back(thread);
        }
        
    }
    else if(expect_num<cur_thread_num)
    {
        vector<gzip_thread*>::iterator thd_ite= _thread_vec.begin();
        for(int i=0;i<cur_thread_num-expect_num;i++)
        {
            (*thd_ite)->Kill_thread();
            DELETE(*thd_ite);
            thd_ite=_thread_vec.erase(thd_ite);    
        }
    }
    return 0; 
}


void compress_unit::In_queue_task(gzip_task& task)
{
    _in_task_num++;
    _waiting_task_queue.push_back(task);
    Process_waiting_task();
}

void compress_unit::Process_waiting_task()
{
    int total_task_num,idle_thd_num,per_thd_task_num,task_shift,thd_index;
    vector<gzip_task>:: iterator beg_task_ite,end_task_ite;
    vector<gzip_task>* tmp_task_vec;
    total_task_num=_waiting_task_queue.size();
    if(total_task_num==0)
        return;
    
    vector<gzip_thread*>  idle_thd_vec;
    vector<gzip_thread*>::iterator thd_ite= _thread_vec.begin();
    
    for(thd_index=0;(unsigned)thd_index<_thread_vec.size();thd_index++)
    {
        if(_thread_vec[thd_index]->Get_waiting_task_queue()==NULL)
        {
            idle_thd_vec.push_back(_thread_vec[thd_index]);    
        }
    }
    
    if((idle_thd_num=idle_thd_vec.size())==0)
        return;
    

    if(total_task_num<idle_thd_num)
    {
            idle_thd_num=total_task_num;
    }
    
    per_thd_task_num=total_task_num/idle_thd_num;
   
    task_shift=0;
    
    for(thd_index=1;thd_index<=idle_thd_num;thd_index++)
    {
        NEW(vector<gzip_task>,tmp_task_vec);
        if( NULL == tmp_task_vec )
        {
            return;
        }
        beg_task_ite=_waiting_task_queue.begin()+task_shift;
      
        if(thd_index!=idle_thd_num)
        {
            end_task_ite=beg_task_ite+per_thd_task_num;
            task_shift+=per_thd_task_num;
        }
        else
        {
            end_task_ite=_waiting_task_queue.end();
        }
        tmp_task_vec->insert(tmp_task_vec->end(), beg_task_ite,end_task_ite);
        idle_thd_vec[thd_index-1]->Set_waiting_task_queue(tmp_task_vec);
        tmp_task_vec=NULL;
    }   
    _waiting_task_queue.clear();
} 
        

void compress_unit::Out_queue_task(vector<gzip_task>& out_queue)
{
    int thread_index;
    int total_thread_num=_thread_vec.size();
    for(thread_index=0;thread_index<total_thread_num;thread_index++)
    {
        vector<gzip_task>* thread_out_queue=_thread_vec[thread_index]->Get_out_task_queue();
        if(thread_out_queue!=NULL)
        {
            out_queue.insert(out_queue.end(),thread_out_queue->begin(),thread_out_queue->end());
            DELETE(thread_out_queue);
        }
    }
   _out_task_num+=out_queue.size();
}



HTTP_SVR_NS_END
